package pro.coman404.tools.table;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author NotFound404 at 2024/7/25
 * code code always code
 */
public class TableDemo2 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        Table table = tableEnv.from(TableDescriptor.forConnector("datagen")
                .option("number-of-rows","10")
                .schema(Schema.newBuilder()
                        .column("uid", DataTypes.TINYINT())
                        .column("payload",DataTypes.STRING())
                        .build())
                .build());

        tableEnv.toDataStream(table)
                .keyBy(r->r.<Byte>getFieldAs("uid"))
                .map(r->"My custom oprator: "+r.<String>getFieldAs("payload"))
                .executeAndCollect().forEachRemaining(System.out::println);

    }

}
